package src.test04;

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.ConfirmCallback;
import com.rabbitmq.client.Connection;
import src.util.ConnectionUtil;

import java.util.UUID;
import java.util.concurrent.ConcurrentNavigableMap;
import java.util.concurrent.ConcurrentSkipListMap;

/**
 * Created with IntelliJ IDEA.
 * Description:
 * User: YKH
 * Date: 2023-08-22
 * Time: 11:50
 */

/**
 * 验证发布确认
 */
public class Task04 {
    public static final int messageConut = 1000;
    public static void main(String[] args) throws Exception{
//        publishMessageIndividually();//单个确认应答, 1000条消息发送单个确认完成,用时26655ms
//        publishMessageMultiply();//1000条消息发送批量确认完成,用时57ms
        publishMessageAsync();

    }

    public static void publishMessageIndividually() throws Exception {
        Connection connection = ConnectionUtil.getConnection();
        Channel channel = connection.createChannel();
        String QUEUE_NAME = UUID.randomUUID().toString();
        channel.queueDeclare(QUEUE_NAME,true,false,false,null);
        //开启发布确认
        channel.confirmSelect();
        long begin = System.currentTimeMillis();
        for(int i = 0; i<messageConut; i++){
            String message = i + "";
            channel.basicPublish("", QUEUE_NAME, null,message.getBytes());
            //客户端确认标识
            boolean flag = channel.waitForConfirms();
            if(flag){
                System.out.println("消息发送成功");
            }
        }
        long end = System.currentTimeMillis();
        System.out.println(messageConut + "条消息发送单个确认完成,用时" + (end - begin) + "ms");
    }

    public static void publishMessageMultiply() throws Exception {
        Connection connection = ConnectionUtil.getConnection();
        Channel channel = connection.createChannel();
        String QUEUE_NAME = UUID.randomUUID().toString();
        channel.queueDeclare(QUEUE_NAME,true,false,false,null);
        //开启发布确认
        channel.confirmSelect();
        //批量确认消息大小
        int batchSize = 10;
        //未确认消息个数
        int outstandingMessageCount = 0;
        long begin = System.currentTimeMillis();
        for(int i = 0; i<messageConut; i++){
            String message = i + "";
            channel.basicPublish("", QUEUE_NAME, null,message.getBytes());
            outstandingMessageCount++;
            if(outstandingMessageCount == batchSize)
                channel.waitForConfirms();
                outstandingMessageCount = 0;
        }
        if(outstandingMessageCount > 0){
            channel.waitForConfirms();
        }
        long end = System.currentTimeMillis();
        System.out.println(messageConut + "条消息发送批量确认完成,用时" + (end - begin) + "ms");
    }

    public static void publishMessageAsync() throws Exception {
        Connection connection = ConnectionUtil.getConnection();
        Channel channel = connection.createChannel();
        String QUEUE_NAME = UUID.randomUUID().toString();
        channel.queueDeclare(QUEUE_NAME,true,false,false,null);
        //开启发布确认
        channel.confirmSelect();
        /**
         * 线程安全有序的一个哈希表，适用于高并发的情况
         * 1.轻松的将序号与消息进行关联
         * 2.轻松批量删除条目 只要给到序列号
         * 3.支持并发访问
         */
        ConcurrentSkipListMap<Long, String> outstandingConfirms = new ConcurrentSkipListMap<>();

        long begin = System.currentTimeMillis();
        //消息确认的回调
        ConfirmCallback ackCallback = (sequenceNumber, multiple) ->{
            //2. 删除已确认的消息, 剩下未确认的消息
            if (multiple) {
                //返回的是小于等于当前序列号的未确认消息 是一个 map
                ConcurrentNavigableMap<Long, String> confirmed =
                        outstandingConfirms.headMap(sequenceNumber, true);
                //清除该部分未确认消息
                confirmed.clear();
            }else{
                //只清除当前序列号的消息
                String message = outstandingConfirms.get(sequenceNumber);
                System.out.println("发布的消息"+message+"已确认，序列号"+sequenceNumber);
                outstandingConfirms.remove(sequenceNumber);
            }
        };
        //未确认消息的回调
        ConfirmCallback failCallback = (sequenceNumber, multiple) ->{
            String message = outstandingConfirms.get(sequenceNumber);
            System.out.println("发布的消息"+message+"未被确认，序列号"+sequenceNumber);
        };
        /**
         * 添加一个异步确认的监听器
         * 1.确认收到消息的回调
         * 2.未收到消息的回调
         */
        channel.addConfirmListener(ackCallback, failCallback);

        for(int i = 0; i<messageConut; i++){
            String message = "消息" + i;
            /**
             * channel.getNextPublishSeqNo()获取下一个消息的序列号
             * 通过序列号与消息体进行一个关联
             * 全部都是未确认的消息体
             */
            //1. 记录下一个发布的消息
            outstandingConfirms.put(channel.getNextPublishSeqNo(), message);
            channel.basicPublish("", QUEUE_NAME, null,message.getBytes());
        }
        long end = System.currentTimeMillis();
        System.out.println(messageConut + "条消息发送单个确认完成,用时" + (end - begin) + "ms");
    }

}
